package com.jungle.boot.mq.controller;// Demo01ProducerTest.java

import com.jungle.boot.mq.producer.*;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.Arrays;
import java.util.List;

/**
 * @author 86189
 * rocketmq测试
 */
@Slf4j
@RestController("/mq")
@RequiredArgsConstructor
public class DemoProducerController {
    private final Demo01Producer producer;
    private final Demo02Producer producer2;
    private final Demo03Producer producer3;
    private final Demo04Producer producer4;
    private final Demo05Producer producer5;

    /**
     * 发送同步请求
     */
    @GetMapping("/testSyncSend")
    public void testSyncSend() throws InterruptedException {
        int id = (int) (System.currentTimeMillis() / 1000);
        SendResult result = producer.syncSend(id);
        log.info("[testSyncSend][发送编号：[{}] 发送结果：[{}]]", id, result);

        // 阻塞等待，保证消费
        // new CountDownLatch(1).await();
    }

    /**
     * 发送异步请求
     */
    @GetMapping("/testASyncSend")
    public void testASyncSend() throws InterruptedException {
        int id = (int) (System.currentTimeMillis() / 1000);
        producer.asyncSend(id, new SendCallback() {

            @Override
            public void onSuccess(SendResult result) {
                log.info("[testASyncSend][发送编号：[{}] 发送成功，结果为：[{}]]", id, result);
            }

            @Override
            public void onException(Throwable e) {
                log.info("[testASyncSend][发送编号：[{}] 发送异常]]", id, e);
            }

        });

        // 阻塞等待，保证消费
        // new CountDownLatch(1).await();
    }

    /**
     * 发送一个消息
     */
    @GetMapping("/testOnewaySend")
    public void testOnewaySend() throws InterruptedException {
        int id = (int) (System.currentTimeMillis() / 1000);
        producer.onewaySend(id);
        log.info("[testOnewaySend][发送编号：[{}] 发送完成]", id);
        // 阻塞等待，保证消费
        // new CountDownLatch(1).await();
    }

    /**
     * 批量发送消息
     */
    @GetMapping("/testSendBatch")
    public void testSendBatch() throws InterruptedException {
        // List<Integer> ids = Stream.iterate(0, i -> i + 1).limit(100).toList();
        List<Integer> ids = Arrays.asList(1, 2, 3);
        SendResult result = producer2.sendBatch(ids);
        log.info("[testSendBatch][发送编号：[{}] 发送结果：[{}]]", ids, result);

    }

    /**
     * 发送定时消息
     */
    @GetMapping("/testSyncSendDelay")
    public void testSyncSendDelay() throws InterruptedException {
        int id = (int) (System.currentTimeMillis() / 1000);
        SendResult result = producer3.syncSendDelay(id, 3); // 延迟级别 3 ，即 10 秒后消费
        log.info("[testSyncSendDelay][发送编号：[{}] 发送结果：[{}]]", id, result);
    }


    /**
     * 失败重试消息发送
     */
    @GetMapping("/testSyncRetrySend")
    public void testSyncRetrySend() throws InterruptedException {
        int id = (int) (System.currentTimeMillis() / 1000);
        SendResult result = producer4.syncSend(id);
        log.info("[testSyncSend][发送编号：[{}] 发送结果：[{}]]", id, result);
    }

    /**
     * 广播消息发送
     */
    public void testBroadcastingSyncSend() throws InterruptedException {
        int id = (int) (System.currentTimeMillis() / 1000);
        SendResult result = producer5.syncSend(id);
        log.info("[testSyncSend][发送编号：[{}] 发送结果：[{}]]", id, result);
    }
}